package caiqr.utils
//
////import com.db.five_million.football_match_sporttery_service
//import caiqr.spark.fb_europe_init.EuropeInitInputFile
//import caiqr.spark.fb_europe_init.EuropeInitSaveDB
//import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, DataFrame}
import org.apache.spark.rdd.RDD
//import java.text.SimpleDateFormat
//import java.sql.DriverManager
//import java.lang.IllegalArgumentException

object PredictionUtils {

  def save_result_to_hdfs(res: RDD[String], output_file: String): Unit ={

    if (!res.isEmpty()) {
      val path = new org.apache.hadoop.fs.Path(output_file);
      val master = output_file.split("9000")(0).concat("9000")
      val hdfs = org.apache.hadoop.fs.FileSystem.get(
        new java.net.URI(master), new org.apache.hadoop.conf.Configuration())
      if (hdfs.exists(path)) {
        hdfs.delete(path, true)
      }
      res.saveAsTextFile(output_file)
    }

  }


  //转换 match_id list
  def transation_match_id_list_to_df(match_id_list_str: String, sc: SparkContext): DataFrame ={
    // 1. Array -> rdd
    val match_id_list_rdd = sc.parallelize(match_id_list_str.split("_"))

    //2. rdd -> DF
    import org.apache.spark.sql.Row
    val new_match_id_list_rdd = match_id_list_rdd.map(p => Row(p))

    import org.apache.spark.sql.types.{StringType, StructField, StructType}
    val newSchemaString = "sporttery_match_id"
    val newSchema =
      StructType(
        newSchemaString.split(",").map(fieldName => StructField(fieldName, StringType)))

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val match_id_list_df = sqlContext.createDataFrame(new_match_id_list_rdd, newSchema)
    match_id_list_df
  }




  //转换 company_id list
  def transation_company_id_list_to_df(company_id_list_str: String, sc: SparkContext): DataFrame ={
    // 1. Array -> rdd
    val company_id_list_rdd = sc.parallelize(company_id_list_str.split("_"))

    //2. rdd -> DF
    import org.apache.spark.sql.Row
    val new_company_id_list_rdd = company_id_list_rdd.map(p => Row(p))

    import org.apache.spark.sql.types.{StringType, StructField, StructType}
    val newSchemaString = "sporttery_company_id"
    val newSchema =
      StructType(
        newSchemaString.split(",").map(fieldName => StructField(fieldName, StringType)))

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val company_id_list_df = sqlContext.createDataFrame(new_company_id_list_rdd, newSchema)
    company_id_list_df
  }






  //转换 tema_id list
  def transation_team_id_list_to_df(team_id_list_str: String, sc: SparkContext): DataFrame ={
    // 1. Array -> rdd
    val team_id_list_rdd = sc.parallelize(team_id_list_str.split("_"))

    //2. rdd -> DF
    import org.apache.spark.sql.Row
    val new_team_id_list_rdd = team_id_list_rdd.map(p => Row(p))

    import org.apache.spark.sql.types.{StringType, StructField, StructType}
    val newSchemaString = "sporttery_team_id"
    val newSchema =
      StructType(
        newSchemaString.split(",").map(fieldName => StructField(fieldName, StringType)))

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val team_id_list_df = sqlContext.createDataFrame(new_team_id_list_rdd, newSchema)
    team_id_list_df
  }


  //计算胜平负预测结果+彩球指数
  def calcultion_goal_result(win: Int, loss: Int) : String = {
    //计算大小球比例
    val result_cnt = win+loss
    val win_ratio = (win.toDouble / result_cnt.toDouble * 100).round.toInt
    val loss_ratio = (loss.toDouble / result_cnt.toDouble * 100).round.toInt


    var final_match_result = ""
    var caiqiu_index = 0
    if (win > loss){
      final_match_result = "1"
      caiqiu_index = win_ratio
    }else{
      final_match_result = "0"
      caiqiu_index = loss_ratio
    }
    s"${final_match_result}_${caiqiu_index}"
  }



    //计算胜平负预测结果+彩球指数
  def calcultion_spf_result(win: Int, draw: Int, loss: Int) : String ={

    //计算胜平负比例
    val result_cnt = win+draw+loss
    val win_ratio = (win.toDouble / result_cnt.toDouble * 100).round.toInt
    val draw_ratio = (draw.toDouble / result_cnt.toDouble * 100).round.toInt
    val loss_ratio = (loss.toDouble / result_cnt.toDouble * 100).round.toInt


    val cnt_map = Map((win, "3"), (draw, "1"), (loss, "0"))
    val cnt = List(win, draw, loss).sorted
    var final_match_result = ""
    var caiqiu_index = 0

    //#查看是否出现相同个数类型
    //#特殊类型:
    //# [11-6-11]
    //# [15-15-5]
    //# [4-11-11]
    //# [10-10-6]
    //# [0-0-4]
    if (cnt_map.size == 2){
      if (win == loss){  // #[11-6-11] [6-11-6] [0-11-0]
        if (win == 0 || win<5){
          final_match_result = "31"
        }else{
          final_match_result = "30"
        }
      }else if (win == draw){
        if (win == 0){
          final_match_result = "0"
        }else if (win > loss) { //#[15-15-5]
          final_match_result = "31"
        }else if (win < loss) { //#[5-5-7]
          final_match_result = "10"
        }
      }else if (draw == loss){
        if (draw == 0 ){ // #[4-0-0]
          final_match_result = "3"
        }else if (draw > win) { //#[4-11-11]
          final_match_result = "10"
        }else if (draw < win) { //#[10-5-5] [20-3-3] [9-3-3]
          if (draw > 4) {
            final_match_result = "31"
          }else if (win > 10) {
            final_match_result = "3"
          }else{
            val cha = win - draw
            if (cha > 5) {
              final_match_result = "3"
            }
            else {
              final_match_result = "31"
            }
          }
        }
      }
    }else {
      if (cnt(2) < 10) {
        if (cnt(2) - cnt(1) > 5)
          final_match_result = cnt_map(cnt(2))
        else {
          final_match_result = cnt_map(cnt(2))
          final_match_result += cnt_map(cnt(1))
        }
      } else {
        final_match_result = cnt_map(cnt(2))
        if (cnt(1) > 4) {
          final_match_result += cnt_map(cnt(1))
        }
      }

      if (final_match_result.length == 1) {
        if (final_match_result == "1") {
          final_match_result = "10"
        }
      }

      if (cnt_map.size == 1) {
        final_match_result = "10"
      }
    }

    // "310" 排序.
    final_match_result = final_match_result.toList.sortWith((a,b)=>a>b).mkString

    //彩球指数
    final_match_result.toString match {
      case "3" => caiqiu_index = win_ratio
      case "1" => caiqiu_index = draw_ratio
      case "0" => caiqiu_index = loss_ratio
      case "31" => caiqiu_index = win_ratio+draw_ratio
      case "30" => caiqiu_index = win_ratio+loss_ratio
      case "10" => caiqiu_index = draw_ratio+loss_ratio
    }

    s"${final_match_result}_${caiqiu_index}"
  }



}


